package com.demo.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.demo.kafka.entity.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

/**
 * kafka消费者
 * @author cyj
 **/
@Slf4j
@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = "test-topic-1", groupId = "group")
    public void listener(KafkaMessage kafkaMessage) {
        log.info("kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    @KafkaListener(topics = "test-topic-1", groupId = "group_partition_0", idIsGroup = false,
            topicPartitions = {
                @TopicPartition(topic = "test-topic-1", partitions = {"0"})
            })
    public void listenerPartiton0(KafkaMessage kafkaMessage) {
        log.info("patition 0 kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    @KafkaListener(topics = "test-topic-1", groupId = "group_partition_0", idIsGroup = false,
            topicPartitions = {
            @TopicPartition(topic = "test-topic-1", partitions = {"1"})
    })
    public void listenerPartiton1(KafkaMessage kafkaMessage) {
        log.info("patition 1 kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    /**
     * 消费延时消息
     * @param msg 消息
     */
    @KafkaListener(topics = "test-topic-2", groupId = "group2")
    public void listenerDelayMsg(String msg) {
        log.info("kafka listenerDelayMsg 消费端消息 : " + msg);
    }
}
